package com.atguigu.edu.app.dws;

import bean.VideoPlayBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import util.DateFormatUtil;
import util.MyKafkaUtil;

import java.time.Duration;

/**
 * 统计各章节的 视频播放次数、累计播放时长、观看人数、
 * //1.基本环境准备
 * //2.检查点相关设置
 * //3.从kafka的播放主题中读取数据
 * //4.转换数据格式 jsonStr->jsonObj
 * //5.指定watermark和时间戳
 * //6.按照tuple2(用户id，视频id)分组
 * //7.开窗（会话窗口）
 * //8.使用处理函数：一个会话窗口是一次视频播放、窗口的所有数据的累计播放时长之和是视频的播放时长，转为实体类
 * //9.按照 用户id 分组
 * //10. 使用状态编程，统计独立播放人数
 * //11.按照章节id分组，
 * //12.聚合，给窗口的开始时间、结束时间、时间戳、观看人数赋值
 * //13.结果写入clickhouse
 */
public class DwsPlayChapterVideoWindow {
    public static void main(String[] args) throws Exception {
        //1.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //2.检查点相关设置
        //3.从kafka的播放主题中读取数据
        String pageLogTopic = "dwd_traffic_video_log";
        String groupId = "dws_play_chapter_video_group";
        DataStreamSource<String> jsonStrDS = env.addSource(MyKafkaUtil.getKafkaConsumer(pageLogTopic, groupId));
        //4.转换数据格式 jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = jsonStrDS.map(JSON::parseObject);
        //5.指定watermark和时间戳
        SingleOutputStreamOperator<JSONObject> withWatermarkDS = jsonObjDS.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject jsonObj, long recordTimestamp) {
                        return jsonObj.getLong("ts");
                    }
                }));
        //6.按照tuple2(设备id，视频id)分组
        KeyedStream<JSONObject, Tuple2<String, String>> keyedDS = withWatermarkDS.keyBy(new KeySelector<JSONObject, Tuple2<String, String>>() {
            @Override
            public Tuple2 getKey(JSONObject jsonObj) throws Exception {
                String mid = jsonObj.getJSONObject("common").getString("mid");
                String videoId = jsonObj.getJSONObject("appVideo").getString("video_id");
                return Tuple2.of(mid, videoId);
            }
        });
        //7.开窗（会话窗口）
        WindowedStream<JSONObject, Tuple2<String, String>, TimeWindow> windowedDS = keyedDS.window(EventTimeSessionWindows.withGap(Time.seconds(30)));
        //8.使用处理函数：一个会话窗口是一次视频播放、窗口的所有数据的累计播放时长之和是视频的播放时长，转为实体类
        SingleOutputStreamOperator<VideoPlayBean> processDS = windowedDS.process(new ProcessWindowFunction<JSONObject, VideoPlayBean, Tuple2<String, String>, TimeWindow>() {
            @Override
            public void process(Tuple2<String, String> tuple2, ProcessWindowFunction<JSONObject, VideoPlayBean, Tuple2<String, String>, TimeWindow>.Context context, Iterable<JSONObject> input, Collector<VideoPlayBean> out) throws Exception {
                VideoPlayBean videoPlayBean = new VideoPlayBean(
                        "",
                        "",
                        "",
                        "",
                        "",
                        "",
                        0L,
                        0L,
                        0L,
                        0L
                );
                for (JSONObject jsonObject : input) {
                    Long durTime = jsonObject.getJSONObject("appVideo").getLong("play_sec");
                    String videoId = jsonObject.getJSONObject("appVideo").getString("video_id");
                    String uid = jsonObject.getJSONObject("common").getString("uid");
                    Long ts = jsonObject.getLong("ts");
                    videoPlayBean.setVideoId(videoId);
                    videoPlayBean.setUid(uid);
                    videoPlayBean.setTs(ts);
                    videoPlayBean.setDuringTime(videoPlayBean.getDuringTime() + durTime);
                }
                videoPlayBean.setPalyCt(1L);
                out.collect(videoPlayBean);
            }
        });
        //9.按照 视频id 分组
        KeyedStream<VideoPlayBean, String> uidKeyedDS = processDS.keyBy(data -> data.getVideoId());
        //10. 使用状态编程，统计视频的独立播放用户
        SingleOutputStreamOperator<VideoPlayBean> withUuCtDS = uidKeyedDS.process(
                new KeyedProcessFunction<String, VideoPlayBean, VideoPlayBean>() {
                    //声明状态,存放用户上次观看该视频的日期
                    private ValueState<String> lastPlayDateState;

                    @Override
                    //给状态赋值，设置生命周期为1天
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("lastPlayDateState", String.class);
                        stateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.days(1)).build());
                        lastPlayDateState = lastPlayDateState = getRuntimeContext().getState(stateDescriptor);
                    }

                    @Override
                    //判断状态中是否为空，或者是否状态中的观看时间，与当前数据的时间不在同一天
                    public void processElement(VideoPlayBean videoPlayBean, KeyedProcessFunction<String, VideoPlayBean, VideoPlayBean>.Context ctx, Collector<VideoPlayBean> out) throws Exception {
                        String curDate = DateFormatUtil.toDate(videoPlayBean.getTs());
                        String lastPlayVideoDate = lastPlayDateState.value();

                        if (StringUtils.isEmpty(lastPlayVideoDate) || !curDate.equals(lastPlayVideoDate)) {
                            videoPlayBean.setUuCt(1L);
                            lastPlayDateState.update(curDate);
                        }
                    }
                }
        );
        //11.维度关联chapter_info

        //11.按照章节id分组，

        //12.聚合，给窗口的开始时间、结束时间、时间戳、观看人数赋值
        //13.结果写入clickhouse
        env.execute();
    }
}
